package com.by.hazelcasttopic;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.by.hazelcasttopic.annotation.AbstractHazelcastTopicListener;
import com.by.hazelcasttopic.annotation.HazelcastTopicListenerContext;
import com.by.hazelcasttopic.consts.HazelcastConst;
import com.by.toolkit.platform.ByContext;
import com.hazelcast.topic.Message;
import com.hazelcast.topic.MessageListener;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;

/**
 * 全局广播消息订阅者统一入口
 *
 * @author YanBingHao
 * @since 2025/3/19
 */
@Slf4j
@Component
public class GlobalHazelcastTopicListener implements MessageListener<String> {

    @Resource
    private HazelcastTopicListenerContext hazelcastTopicListenerContext;

    @Override
    public void onMessage(Message<String> message) {
        log.info("[GlobalHazelcastTopicListener][onMessage] message:[{}]", JSON.toJSONString(message));
        String messageObject = message.getMessageObject();
        JSONObject jsonObject = JSONObject.parseObject(messageObject);
        JSONObject context = jsonObject.getJSONObject(HazelcastConst.BY_CONTEXT);
        String topicName = jsonObject.getString(HazelcastConst.TOPIC_NAME);
        if (Objects.nonNull(context) && !context.isEmpty()) {
            context.keySet()
                    .forEach(key -> {
                        String contextVal = context.getString(key);
                        log.info("[GlobalHazelcastTopicListener][onMessage] key:[{}] contextVal:[{}]", key, contextVal);
                        ByContext.put(key, contextVal);
                    });
            Map<String, String> allContext = ByContext.getAll();
            log.info("[GlobalHazelcastTopicListener][onMessage] allContext:[{}]", allContext);
        }

        if (StringUtils.isBlank(topicName)) {
            log.warn("[GlobalHazelcastTopicListener][onMessage] topicNameIsNull");
            return;
        }
        Class<?> argClass = hazelcastTopicListenerContext.getArgClass(topicName);
        AbstractHazelcastTopicListener listener = hazelcastTopicListenerContext.get(topicName);
        if (Objects.isNull(listener)) {
            log.info("[GlobalHazelcastTopicListener][onMessage] topicName:[{}] listenerIsNull", topicName);
            return;
        }
        log.info("[GlobalHazelcastTopicListener][onMessage] dealStart topicName:[{}]", topicName);
        listener.process(jsonObject.toJavaObject(argClass));
        log.info("[GlobalHazelcastTopicListener][onMessage] dealEnd topicName:[{}]", topicName);
    }

}
